package ds_industry_2025.industry.gy_04.T5

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.classification.RandomForestClassifier
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.ml.tuning.{CrossValidator, ParamGridBuilder}
import org.apache.spark.sql.SparkSession

import java.util.Properties

/*
      1、根据子任务一的结果，建立随机森林（随机森林相关参数可自定义，不做限制），使用子任务一的结果训练随机森林模型，然后再将hudi
      中dwd.fact_machine_learning_data_test（该表字段含义与dwd.fact_machine_learning_data表相同，
      machine_record_state列值为空，表结构自行查看）转成向量，预测其是否报警将结果输出到MySQL数据库shtd_industry中
      的ml_result表中（表结构如下）。在Linux的MySQL命令行中查询出machine_record_id为1、8、20、28和36的5条数据，将SQL语句
      制并粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下，将执行结果截图粘贴至客户端桌面【Release\任务C提交
      结果.docx】中对应的任务序号下。
ml_result表结构：
字段	类型	中文含义	备注
machine_record_id	int	主键
machine_record_state	double	设备状态	报警为1，其他状态则为0
 */
object t1 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t1")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._

    //  todo 读取特征工程的结果
    //  todo 其中下面更改字段名，这一步非常重要，因为在监督学习中，模型需要知道哪个字段是目标变量(即我们要预测的变量)
    //  todo 这个目标变量通常被称为标签(label)
    val data = spark.table("dwd.fact_machine_learding_data")
      .withColumnRenamed("machine_record_state","label")

    //  todo 将数据的特征列全部提取出来(xml文档里面解析出来的几列)
    val cols = data.columns.slice(3, data.columns.length - 5)

    //  todo 创建将特征字段转化为特征向量
    val assembler = new VectorAssembler()
      .setInputCols(cols)
      .setOutputCol("features")


    //  todo 创建一个随机森林分类器
    val classifier = new RandomForestClassifier()

    //  todo 创建一个机器学习流水线
    val pipeline = new Pipeline()
      .setStages(Array(assembler,classifier))

    //  todo  创建一个超参数网格 来决策出最优随机森林的参数
    //  classifier.impurity: 用于衡量不纯度的标准，这里使用"entropy"（信息增益）和"gini"（基尼不纯度）。
//   classifier.maxDepth: 每棵树的最大深度，尝试深度为5、10和15。
    val paramMpas = new ParamGridBuilder()
      .addGrid(classifier.numTrees,Array(10,20,30))
      .addGrid(classifier.maxDepth,Array(5,10,15))
      .addGrid(classifier.impurity,Array("gini"))
      .build()

    //  todo 使用交叉验证的形式      validator:验证器   Evaluator:评估者
    //  setEstimatorParamMaps(paramMaps): 设置要测试的参数组合，即在第一步构建的paramMaps。
    //   setEvaluator(new MulticlassClassificationEvaluator()): 设置评估器，这里使用多类分类评估器，评估模型的分类性能（如准确率、召回率等）
    //   setEstimator(pipeline): 设置要训练的模型管道，这个管道包括特征提取和分类器。
    val crossValidator = new CrossValidator()
      .setEstimatorParamMaps(paramMpas)
      .setEvaluator(new MulticlassClassificationEvaluator())
      .setEstimator(pipeline)
      .setParallelism(4)

    //  todo 使用交叉验证训练模型
    val module = crossValidator.fit(data)

    //  todo 下面可以进行读取预测的数据进行预测    其中预测完的字段名称为prediction


    val Array(rading,test) = data.randomSplit(Array(0.8, 0.2))

    //  todo 对数据进行预测
    val result = module.transform(test).select("machine_record_id","prediction")

   val conn=new Properties()
   conn.setProperty("user","root")
    conn.setProperty("password","123456")
    conn.setProperty("driver","com.mysql.jdbc.Driver")

    result.write.mode("overwrite")
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_industry?useSSL=false","ml_result",conn)


    //  select * from ml_result where machine_record_id in(14747628,14747632,14747645,14747630,14748618,14748623);



    spark.close()
  }

}
